/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.streaming.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.api.java.JavaRDD;
import org.ipig.commons.helper.GsonHelper;
import org.ipig.commons.helper.StringHelper;
import org.ipig.computing.spark.SparkLauncherService;
import org.ipig.computing.spark.conf.SparkExecutorConf;
import org.ipig.computing.spark.streaming.SparkStreamExecutorLauncher;

import java.util.Iterator;

/**
 * SparkStreamKafkaToSpikeExecutorService
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
public interface SparkStreamKafkaToSpikeExecutorService extends SparkStreamKafkaExecutorService, SparkLauncherService<SparkExecutorConf> {
    @Override
    public default void initEachPartition(Iterator<ConsumerRecord<String, String>> partition) {
//        System.out.println(Thread.currentThread().getName() + "_" + Thread.currentThread().getId() + "  " + "initEachPartition");
    }

    @Override
    public default void postEachPartition(Iterator<ConsumerRecord<String, String>> partition) {

    }

    @Override
    public default String getMainClass() {
        return SparkStreamExecutorLauncher.class.getName();
    }

    @Override
    public default String getMainArg() {
        String json = "";
        SparkExecutorConf executorConf = this.getConf();
        executorConf = executorConf.deepClone();
        //@TODO 开发使用，上测试与生产前必须关闭
//        String group = RandomStringUtils.randomAlphanumeric(10);
//        executorConf.getSparkKafkaConf().getConsumerConf().setGroupId(group);
        executorConf.setExecutorClass(this.getExecutorClass());
        json = GsonHelper.toJson(executorConf, SparkExecutorConf.class);
//        System.out.println(json);
        return StringHelper.encodeByUTF8(json);
    }

    @Override
    public default String getExecutorClass() {
        return this.getClass().getName();
    }

    @Override
    public default void postEachRDD(JavaRDD<ConsumerRecord<String, String>> rdd) {
//        System.out.println(Thread.currentThread().getName() + "_" + Thread.currentThread().getId() + " 可以进行累加器值的获取");
    }
}

